#include "vzconn/conn/tcpconnect.h"
#include "vzconn/pool/perfectbufferpool.h"

namespace vzconn {

////////////////////////////////////////////////////////////////////////////////

TcpConnect::TcpConnect(boost::asio::io_service &io_service)
  : VzConnect(io_service) {
  socket_.reset(new boost::asio::ip::tcp::socket(io_service));
}

TcpConnect::~TcpConnect() {
  LOG(WARNING) << "~TcpConnect()";
}

bool TcpConnect::Start() {
  if(state_ == CS_CLOSED && socket_->is_open()) {
    state_ = CS_CONNECTED;
  } else {
    return false;
  }
  if(!EnableNagleAlgorithm(false)) {
    return false;
  }
  remote_address_ = socket_->remote_endpoint().address().to_string();
  io_service_.post(
    boost::bind(&VzConnect::AsyncReadData, shared_from_this()));
  return true;
}

bool TcpConnect::EnableNagleAlgorithm(bool is_enable) {
  if (state_ == CS_CLOSED) {
    LOG(WARNING) << "The state is CS_CLOSED";
    return false;
  }
  boost::asio::ip::tcp::no_delay option(is_enable);
  socket_->set_option(option);
  return true;
}

void TcpConnect::AsyncReadData() {
  if (state_ == CS_CLOSED) {
    LOG(WARNING) << "The socket is closed";
    return;
  }
  socket_->async_read_some(
    boost::asio::buffer(pre_read_buffer_, PRE_BUFFER_SIZE),
    boost::bind(&VzConnect::HandleReadData, shared_from_this(),
                boost::asio::placeholders::error,
                boost::asio::placeholders::bytes_transferred));
}

void TcpConnect::TryToWriteData() {

  if (is_writing_) {
    DLOG(WARNING) << "Has data wrting";
    return;
  }
  if(buffers_.size() == 0) {
    return;
  }
  is_writing_ = true;
  ByteBuffer::Ptr write_buffer = buffers_.front();
  socket_->async_write_some(
    boost::asio::buffer(write_buffer->Data(), write_buffer->size()),
    boost::bind(&VzConnect::HandleDataWrite, shared_from_this(),
                boost::asio::placeholders::error,
                write_buffer,
                boost::asio::placeholders::bytes_transferred));
}

void TcpConnect::ConnectError(const boost::system::error_code& err) {
  if (state_ == CS_CLOSED) {
    LOG(INFO) << "state == CS_CLOSED then return";
    return;
  }
  state_ = CS_CLOSED;
  if (socket_->is_open()) {
    LOG(INFO) << "sockset is closed";
    socket_->close();
  }
  LOG(INFO) << "Signal to up layer";
  SignalConnectError(shared_from_this(), err);
}

void TcpConnect::HandleCloseConnect() {
  BOOST_ASSERT(state_ == CS_CLOSED);
  if (socket_->is_open()) {
    socket_->close();
  }
}

////////////////////////////////////////////////////////////////////////////////

BlockVzTcpConnect::BlockVzTcpConnect(boost::asio::io_service &io_service)
  : BlockVzConnect(io_service),
    socket_(io_service) {
}

BlockVzTcpConnect::~BlockVzTcpConnect() {
}

void BlockVzTcpConnect::CloseConnect() {
  if(socket_.is_open()) {
    socket_.close();
  }
}

bool BlockVzTcpConnect::RealySyncWrite(
  std::vector<boost::asio::const_buffer> &buffers) {
  if(!socket_.is_open()) {
    LOG(ERROR) << "The socket is not opened";
    return false;
  }
  try {
    boost::asio::write(socket_, buffers);
  } catch(std::exception &e) {
    LOG(ERROR) << e.what();
    socket_.close();
    return false;
  }
  return true;
}

bool BlockVzTcpConnect::RealySyncRead(char *buffer, uint32 buffer_size) {
  if(!socket_.is_open()) {
    LOG(ERROR) << "The socket is not opened";
    return false;
  }
  uint32 read_size = 0;
  try {
    read_size = boost::asio::read(
                  socket_,
                  boost::asio::buffer(buffer, buffer_size));
  } catch(std::exception &e) {
    LOG(ERROR) << e.what();
    socket_.close();
    return false;
  }
  return read_size == buffer_size;
}


}